/* This file is part of BonQ - A simple HTTP-based messaging server.
 * Copyright (C) 2009 James Cook
 *
 * BonQ is free software: you can redistribute it and/or modify it under the
 * terms of the GNU General Public License, version 2, as published by the
 * Free Software Foundation.
 *
 * BonQ is distributed in the hope that it will be useful, but WITHOUT ANY
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE.  See the GNU General Public License, version 2,
 * for more details.
 *
 * You should have received a copy of the GNU General Public License, version
 * 2, along with BonQ.  If not, see:
 *
 * http://www.gnu.org/licenses/old-licenses/gpl-2.0.html
 */
package com.bonkabonka.queue;

import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;

/**
 * This queue will remove itself after a specified period of inactivity.  As it
 * is an extension of TransientQueue, no messages will be saved if the server
 * is downed.
 * 
 * @author jamesc
 */
public class TemporaryQueue extends TransientQueue
{
    private static final int PERIOD = 60 * 1000; // one minute

    private final transient Object mutex = new Object();
    private transient long activity;
    private long timeout = 60 * PERIOD;
    
    /**
     * Thread to rummage through registered queues and zap them if they've been
     * idle too long.
     */
    private static class Watcher extends Thread
    {
        private final HashMap<QueueBucket, LinkedList<TemporaryQueue>> buckets =
                new HashMap<QueueBucket, LinkedList<TemporaryQueue>>();
        
        /**
         * Sleep for a while and call expire upon waking.  Do it until
         * interrupted.
         */
        @Override
        public void run()
        {
            while(true)
            {
                try
                {
                    sleep(PERIOD);
                    
                    synchronized(buckets)
                    {
                        expire();
                    }
                }
                catch(InterruptedException ie)
                {
                    ie.printStackTrace(System.err);
                    break;
                }
            }
        }
        
        /**
         * This expires queues when they're past their timeout.  Icky code that
         * makes my brain hurt with all its nested Iterators.  It looked so
         * much nicer as foreach constructs.
         *
         * Of course, when I wrote that comment, I neglected to mention WHY I
         * changed the code.  I assume it had to do with concurrent modification
         * exceptions, but I really don't remember.
         */
        private void expire()
        {
            long now = System.currentTimeMillis();
            
            Iterator<LinkedList<TemporaryQueue>> queueListIterator =
                    buckets.values().iterator();
            while(queueListIterator.hasNext())
            {
                LinkedList<TemporaryQueue> queueList = queueListIterator.next();
                
                Iterator<TemporaryQueue> queueIterator = queueList.iterator();
                while(queueIterator.hasNext())
                {
                    TemporaryQueue q = queueIterator.next();
                    if(q.activity + q.timeout < now)
                    {
                        q.getBucket().remove(q.getName());
                        queueIterator.remove();
                    }
                }
                
                if(queueList.size() == 0)
                {
                    queueListIterator.remove();
                }
            }
        }
        
        /**
         * Reguster the given queue with the watcher.
         * 
         * @param queue TemporaryQueue to watch
         */
        public void add(TemporaryQueue queue)
        {
            QueueBucket bucket = queue.getBucket();

            synchronized(buckets)
            {
                LinkedList<TemporaryQueue> queues = buckets.get(bucket);
                if(queues == null)
                {
                    queues = new LinkedList<TemporaryQueue>();
                    buckets.put(bucket, queues);
                }
                
                queues.add(queue);
            }
        }
    }
    
    /**
     * All instances of the class in the same classloader share the same watcher
     */
    private static final Watcher watcher = new Watcher();
    
    /**
     * Start the watcher when the class is loaded
     */
    static
    {
        watcher.setName("TemporaryQueue Watcher");
        watcher.start();
    }
    
    /**
     * Initialize the FIFO and register it with the watcher thread.
     * 
     * @param bucket QueueBucket that owns the queue
     * @param name String containing the name of the queue
     */
    @Override
    public void init(QueueBucket bucket, String name)
    {
        super.init(bucket, name);
        
        activity = System.currentTimeMillis();
        
        watcher.add(this);
    }
    
    /**
     * Adjust the timeout of the queue.  Pass in a zero or an invalid string to
     * have the current value returned.
     * 
     * @param message String containing the new timeout value
     * @return String containing the timeout value
     */
    @Override
    public String configure(String message)
    {
        synchronized(mutex)
        {
            activity = System.currentTimeMillis();
        }

        message = message.trim();
        long value = timeout / PERIOD;
        
        try
        {
            value = Long.parseLong(message);
            if(value > 0L)
            {
                timeout = value * PERIOD;
            }
        }
        catch(NumberFormatException nfe)
        {
            nfe.printStackTrace(System.err);
        }
        
        return Long.toString(value);
    }
    
    /**
     * Add a message to the queue.
     * 
     * @param message String containing the message
     * @return null;
     */
    @Override
    public String push(String message)
    {
        synchronized(mutex)
        {
            activity = System.currentTimeMillis();
        }
        
        return super.push(message);
    }
    
    /**
     * Return the next message in the queue, but don't remove it.
     * 
     * @return String containing the message
     */
    @Override
    public String peek()
    {
        synchronized(mutex)
        {
            activity = System.currentTimeMillis();
        }
        
        return super.peek();
    }
    
    /**
     * Return the next message in the queue.
     * 
     * @return String containing the message
     */
    @Override
    public String pull()
    {
        synchronized(mutex)
        {
            activity = System.currentTimeMillis();
        }
        
        return super.pull();
    }
}
